package com.navinfo.opentsp.platform.computing.selector;

import org.apache.flume.Channel;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.channel.AbstractChannelSelector;

import java.util.ArrayList;
import java.util.List;

/**
 * 轮询选择Channel的Seletor<br/>
 *
 * <B>背景:</B> Kafka Sink 单线程同步调用，吞吐无法继续提升。
 * <B>解决方案：</B>自定义实现channel selector, 将source的流量均匀分发到多个channel, 并让每个channel由一个独立的kafka sink消费<br/>
 * <B>配置：</B>为source指定selector.type=com.navinfo.opentsp.platform.computing.selector.RRChannelSelector<br/>
 */
public class RRChannelSelector extends AbstractChannelSelector {
    private static final List<Channel> EMPTY_LIST = new ArrayList<Channel>();
    private int rrIndex = 0;

    public List<Channel> getRequiredChannels(Event event) {
        List<Channel> allChannels = getAllChannels();

        List<Channel> result = new ArrayList<Channel>();
        result.add(allChannels.get(rrIndex));

        rrIndex = (rrIndex + 1) % allChannels.size();
        return result;
    }

    public List<Channel> getOptionalChannels(Event event) {
        return RRChannelSelector.EMPTY_LIST;
    }

    public void configure(Context context) { }
}